###############################################################################
#                                                                             #
#    This file is part of the testsuite of the "ssbot" Python module.         #
#    It is distributed under the MIT License. See LICENSE.txt for details.    #
#                                                                             #
###############################################################################


import os
import os.path
import shutil
import sqlite3
import sys
import unittest

sys.path.append('../')

import ssbot.ssdb
import ssbot.ssutils


class BuilderDBTests(unittest.TestCase):
    def setUp(self):
        self.test_builder_name = 'test_builder'
        os.mkdir(self.test_builder_name)
        self.db_file = os.path.join(
            '.', self.test_builder_name, '%s.db' % self.test_builder_name)
        self.db = ssbot.ssdb.BuilderDB(self.test_builder_name)

    def tearDown(self):
        if os.path.exists(self.test_builder_name):
            shutil.rmtree(self.test_builder_name)

    def test_db_file_creation(self):
        self.assertTrue(os.path.exists(os.path.join(
            '.', self.test_builder_name, '%s.db' % self.test_builder_name)))


    def test_add_new_build(self):
        self.db.add_new_build(1, 'test_add_new_build')
        conn = sqlite3.connect(self.db_file)
        c = conn.cursor()
        c.execute('SELECT * FROM %s' % ssbot.ssdb.BUILDS_TABLE_NAME)
        all_builds = []
        while True:
            build = c.fetchone()
            if build:
                all_builds.append(build)
            else:
                break
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILDS_TABLE_NAME, ssbot.ssutils.STATUS_PENDING))
        running_builds = []
        while True:
            build = c.fetchone()
            if build:
                running_builds.append(build)
            else:
                break
        conn.close()
        self.assertEqual(1, len(all_builds))
        self.assertEqual(1, len(running_builds))

    def test_start_build(self):
        self.db.add_new_build(1, 'test_start_build')
        self.db.start_build(1, 'start_time')
        conn = sqlite3.connect(self.db_file)
        c = conn.cursor()
        c.execute('SELECT * FROM %s' % ssbot.ssdb.BUILDS_TABLE_NAME)
        builds = []
        while True:
            build = c.fetchone()
            if build:
                builds.append(build)
            else:
                break
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILDS_TABLE_NAME, ssbot.ssutils.STATUS_PENDING))
        pending_builds = []
        while True:
            build = c.fetchone()
            if build:
                pending_builds.append(build)
            else:
                break
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILDS_TABLE_NAME, ssbot.ssutils.STATUS_RUNNING))
        running_builds = []
        while True:
            build = c.fetchone()
            if build:
                running_builds.append(build)
            else:
                break
        conn.close()
        self.assertEqual(1, len(builds))
        self.assertEqual(0, len(pending_builds))
        self.assertEqual(1, len(running_builds))

    def test_finish_build(self):
        self.db.add_new_build(1, 'test_finish_build')
        self.db.start_build(1, 'start_time')
        self.db.finish_build(1, 'pass', 'end_time')
        conn = sqlite3.connect(self.db_file)
        c = conn.cursor()
        c.execute('SELECT * FROM %s' % ssbot.ssdb.BUILDS_TABLE_NAME)
        builds = []
        while True:
            build = c.fetchone()
            if build:
                builds.append(build)
            else:
                break
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILDS_TABLE_NAME, ssbot.ssutils.STATUS_PENDING))
        pending_builds = []
        while True:
            build = c.fetchone()
            if build:
                pending_builds.append(build)
            else:
                break
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILDS_TABLE_NAME, ssbot.ssutils.STATUS_RUNNING))
        running_builds = []
        while True:
            build = c.fetchone()
            if build:
                running_builds.append(build)
            else:
                break
        c.execute(
            'SELECT * FROM %s WHERE status="%s"' %
            (ssbot.ssdb.BUILDS_TABLE_NAME, ssbot.ssutils.STATUS_COMPLETED))
        completed_builds = []
        while True:
            build = c.fetchone()
            if build:
                completed_builds.append(build)
            else:
                break
        conn.close()
        self.assertEqual(1, len(builds))
        self.assertEqual(0, len(pending_builds))
        self.assertEqual(0, len(running_builds))
        self.assertEqual(1, len(completed_builds))

    def test_get_build_count(self):
        self.assertEqual(0, self.db.get_build_count())
        self.db.add_new_build(1, 'test_get_completed_build_count')
        self.assertEqual(1, self.db.get_build_count())
        self.db.finish_build(1, 'pass', 'end_time')
        self.assertEqual(1, self.db.get_build_count())

    def test_add_step(self):
        self.db.add_new_build(1, 'test_add_step')
        self.db.add_step(1, 'new_step', 'start_time')
        conn = sqlite3.connect(self.db_file)
        c = conn.cursor()
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILD_STEPS_TABLE_NAME_FORMAT % 1,
                   ssbot.ssutils.STATUS_RUNNING))
        running_steps = []
        while True:
            build = c.fetchone()
            if build:
                running_steps.append(build)
            else:
                break
        conn.close()
        self.assertEqual(1, len(running_steps))

    def test_finish_step(self):
        self.db.add_new_build(1, 'test_finish_step')
        self.db.add_step(1, 'new_step', 'start_time')
        self.db.finish_step(1, 'new_step', 'pass', 'end_time')
        conn = sqlite3.connect(self.db_file)
        c = conn.cursor()
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILD_STEPS_TABLE_NAME_FORMAT % 1,
                   ssbot.ssutils.STATUS_RUNNING))
        running_steps = []
        while True:
            build = c.fetchone()
            if build:
                running_steps.append(build)
            else:
                break
        c.execute('SELECT * FROM %s WHERE status="%s"' %
                  (ssbot.ssdb.BUILD_STEPS_TABLE_NAME_FORMAT % 1,
                   ssbot.ssutils.STATUS_COMPLETED))
        completed_steps = []
        while True:
            build = c.fetchone()
            if build:
                completed_steps.append(build)
            else:
                break
        conn.close()
        self.assertEqual(0, len(running_steps))
        self.assertEqual(1, len(completed_steps))

    def test_get_build(self):
        self.db.add_new_build(1, 'test_get_build1')
        self.db.start_build(1, 'start_time1')
        self.db.finish_build(1, ssbot.ssutils.RESULT_PASS, 'end_time')
        self.db.add_new_build(2, 'test_get_build2')
        self.db.start_build(2, 'start_time2')
        self.db.add_new_build(3, 'test_get_build3')
        build1 = self.db.get_build(1)
        self.assertEqual(build1['build_number'], 1)
        self.assertEqual(build1['reason'], 'test_get_build1')
        self.assertEqual(build1['status'], ssbot.ssutils.STATUS_COMPLETED)
        self.assertEqual(build1['result'], ssbot.ssutils.RESULT_PASS)
        build2 = self.db.get_build(2)
        self.assertEqual(build2['build_number'], 2)
        self.assertEqual(build2['reason'], 'test_get_build2')
        self.assertEqual(build2['status'], ssbot.ssutils.STATUS_RUNNING)
        build3 = self.db.get_build(3)
        self.assertEqual(build3['build_number'], 3)
        self.assertEqual(build3['reason'], 'test_get_build3')
        self.assertEqual(build3['status'], ssbot.ssutils.STATUS_PENDING)
        build4 = self.db.get_build(4)
        self.assertEqual(build4, None)

    def test_get_all_steps(self):
        self.db.add_new_build(1, 'test_get_all_steps1')
        self.db.add_step(1, 'step1', 'start_time1')
        self.db.finish_step(1, 'step1', ssbot.ssutils.RESULT_PASS, 'end_time1')
        self.db.add_step(1, 'step2', 'start_time2')
        self.db.add_new_build(2, 'test_get_all_steps2')
        steps = self.db.get_all_steps(1)
        self.assertEqual(len(steps), 2)
        step1 = steps[0]
        self.assertEqual(step1['name'], 'step1')
        self.assertEqual(step1['status'], ssbot.ssutils.STATUS_COMPLETED)
        self.assertEqual(step1['result'], ssbot.ssutils.RESULT_PASS)
        step2 = steps[1]
        steps = self.db.get_all_steps(2)
        self.assertEqual(len(steps), 0)
        steps = self.db.get_all_steps(3)
        self.assertEqual(steps, None)

    def test_get_step(self):
        self.db.add_new_build(1, 'test_get_all_steps1')
        self.db.add_step(1, 'step1', 'start_time1')
        self.db.finish_step(1, 'step1', ssbot.ssutils.RESULT_PASS, 'end_time1')
        step = self.db.get_step(1, 'step1')
        self.assertNotEqual(step, None)
        self.assertEqual(step['name'], 'step1')
        step = self.db.get_step(1, 'step2')
        self.assertEqual(step, None)


if __name__ == '__main__':
    unittest.main()
